package com.xiangxiao.rpan.data.partitioner;

import com.alibaba.fastjson.JSON;
import com.xiangxiao.rpan.data.constant.GlobalConsts;
import com.xiangxiao.rpan.data.entity.KafkaAlarm;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;

/**
 * @auther xiangxiao
 * @email 573768011@qq.com
 * @data 2024/7/17 11:19
 * 1.实现接口Partitioner
 * 2.实现3个方法:partition，close，configure
 * 3.编写partition方法，返回分区号
 */
public class RpandataPartitioner implements Partitioner {
  /**
   * 返回信息对应的翻去
   * @param topic         主题
   * @param key           消息的key
   * @param keyBytes      消息的key序列化后的字节数组
   * @param value         消息的value
   * @param valueBytes    消息的value序列化后的字节数组
   * @param cluster       集群元数据可以查看分区信息
   * @return
   */
  @Override
  public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
//    // 获取消息
    String msgValue = new String(valueBytes, StandardCharsets.UTF_8);
    // 创建partition
    int partitionNum = 0;
    // 判断消息是否包含test
    KafkaAlarm alarm = JSON.parseObject(msgValue, KafkaAlarm.class);
    Date alarmTime = alarm.getTime();
    // 做一个按告警时间的秒值进行分区的策略
    Integer seconds = alarmTime.getSeconds();
    if (seconds < 15){
      partitionNum = 0;
    } else if (seconds >= 15 && seconds < 30) {
      partitionNum = 1;
    } else if (seconds >= 30 && seconds < 45) {
      partitionNum = 2;
    } else if (seconds >= 45) {
      partitionNum = 3;
    }
    // 返回分区号
    System.out.println("分区号:" + partitionNum);
    return partitionNum;
  }

  @Override
  public void close() {

  }

  @Override
  public void configure(Map<String, ?> map) {

  }
}
